#include "rtc_worker.h"
#include "signaling_worker.h"

#include <unistd.h>
#include <rtc_base/logging.h>



namespace xrtc {

void rtcWorkerRecvNotify(EventLoop *, IOWatcher *, int fd, int, void *data) {
    int msg;
    if (read(fd, &msg, sizeof(int)) != sizeof(int)) {
        RTC_LOG(LS_WARNING) << "read from pipe error: " << strerror(errno) << ", errno: " << errno;
        return;
    }

    RtcWorker* worker = (RtcWorker*)data;
    worker->_processNotify(msg);
}

RtcWorker::RtcWorker(int worker_id, const RtcServerOptions& options) 
         : m_options(options), m_worker_id(worker_id), m_el(new EventLoop(this))
         , m_rtc_stream_manager(std::make_unique<RtcStreamManager>(m_el))  {
                                                                        
}

RtcWorker::~RtcWorker() {
    if (m_el) {
        delete m_el;
        m_el = nullptr;
    }

    if (m_thread) {
        m_thread.reset();
    }
}

int RtcWorker::init() {
    int fds[2];
    if (pipe(fds)) {
        RTC_LOG(LS_WARNING) << "create pipe error: " << strerror(errno) << ", errno: " << errno;
        return -1;
    }
    
    m_notify_recv_fd = fds[0];
    m_notify_send_fd = fds[1];

    m_pipe_watcher = m_el->createIOEvent(rtcWorkerRecvNotify, this);
    m_el->startIOEvent(m_pipe_watcher, m_notify_recv_fd, EventLoop::READ);

    return 0;
}

bool RtcWorker::start() {
    if (m_thread) {
        RTC_LOG(LS_WARNING) << "rtc worker already start, worker_id: " << m_worker_id;
        return false;
    }
    
    m_thread = std::make_unique<std::thread> ([=]() {
        RTC_LOG(LS_INFO) << "rtc worker event loop start, worker_id: " << m_worker_id;
        m_el->start();
        RTC_LOG(LS_INFO) << "rtc worker event loop stop, worker_id: " << m_worker_id;
    });

    return true;
}

void RtcWorker::stop() {
    notify(QUIT);
}

int RtcWorker::notify(int msg) {
    int written = write(m_notify_send_fd, &msg, sizeof(int));
    return written == sizeof(int) ? 0 : -1;
}

void RtcWorker::join() {
    if (m_thread && m_thread->joinable()) {
        m_thread->join();
    }
}

void RtcWorker::_stop() {
    if (!m_thread) {
        RTC_LOG(LS_WARNING) << "rtc worker not running, worker_id: " << m_worker_id;
        return;
    }

    m_el->deleteIOEvent(m_pipe_watcher);
    m_el->stop();
    close(m_notify_recv_fd);
    close(m_notify_send_fd);
}

void RtcWorker::pushMsg(std::shared_ptr<RtcMsg> msg) {
    m_msg_queue.produce(msg);
}

bool RtcWorker::popMsg(std::shared_ptr<RtcMsg>* msg) {
    return m_msg_queue.consume(msg);
}

int RtcWorker::sendRtcMsg(std::shared_ptr<RtcMsg> msg) {
    // 将消息投递到worker的队列
    pushMsg(msg);
    return notify(RTC_MSG);
}

// 处理不同类型的消息
void RtcWorker::_processNotify(int msg) {
    switch (msg) {
        case QUIT:
            _stop();
            break;
        case RTC_MSG:
            _processRtcMsg();
            break;
        default:
            RTC_LOG(LS_WARNING) << "unknown msg: " << msg;
            break;
    }
}

// 解析命令参数
void RtcWorker::_processRtcMsg() {
    std::shared_ptr<RtcMsg> msg;
    if (!popMsg(&msg)) {
        return;
    }

    RTC_LOG(LS_INFO) << "cmdno[" << msg->m_cmdno << "] uid[" << msg->m_uid
        << "] stream_name[" << msg->m_stream_name << "] audio[" << msg->m_audio
        << "] video[" << msg->m_video << "] log_id[" << msg->m_log_id
        << "] rtc worker receive msg, worker_id: " 
        << m_worker_id;

    switch (msg->m_cmdno) {
        case CMDNO_PUSH:
            _process_push(msg);
            break;
        case CMDNO_ANSWER:
            _process_answer(msg);
            break;
        default:
            RTC_LOG(LS_WARNING) << "unknown cmdno: " << msg->m_cmdno << ", log_id: " << msg->m_log_id;
            break;
    }
}

// 向信令服务器线程返回offer
void RtcWorker::_process_push(std::shared_ptr<RtcMsg> msg) {
    std::string offer;

    int ret = m_rtc_stream_manager->createPushStream(msg->m_uid, msg->m_stream_name, msg->m_audio, 
                        msg->m_video, msg->m_log_id, (rtc::RTCCertificate*)(msg->m_certificate), offer);
    if (ret != 0) {
        RTC_LOG(LS_ERROR) << "create push stream error, ret: " << ret << ", log_id: " << msg->m_log_id; 
    }                         

    RTC_LOG(LS_INFO) << "offer: " << offer;

    msg->m_sdp = offer;

    SignalingWorker* worker = (SignalingWorker*)msg->m_worker;
    if (worker) {
        worker->sendRtcMsg(msg);
    }
}

void RtcWorker::_process_answer(std::shared_ptr<RtcMsg> msg) {
    int ret = m_rtc_stream_manager->set_answer(msg->m_uid, msg->m_stream_name,
            msg->m_sdp, msg->m_stream_type, msg->m_log_id);
    
    
    RTC_LOG(LS_INFO) << "rtc worker process answer, uid: " << msg->m_uid
        << ", stream_name: " << msg->m_stream_name
        << ", worker_id: " << m_worker_id
        << ", log_id: " << msg->m_log_id
        << ", ret: " << ret;
}

}